package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.Collection;
import java.util.Set;

/**
 * @author Felix
 * @date 2023/3/11
 * 将流中的数据写到phoenix表中
 */
public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //jsonObj : {"tm_name":"三星7","sink_table":"dim_base_trademark","id":1}
        //upsert into 库.表 (a,b,c,d) values('aa','bb','cc','dd')
        String tableName = jsonObj.getString("sink_table");
        //{"tm_name":"三星7","id":1}
        jsonObj.remove("sink_table");

        String type = jsonObj.getString("type");
        jsonObj.remove("type");

        Set<String> keys = jsonObj.keySet();

        Collection<Object> values = jsonObj.values();

        //拼接upsert语句
        String upsertSql = "upsert into " + GmallConfig.PHOENIX_SCHEMA + "." + tableName
            + " (" + StringUtils.join(keys, ",")
            + ") values('" + StringUtils.join(values, "','") + "')";

        System.out.println("在phoenix表中执行的upsert语句:" + upsertSql);

        PhoenixUtil.executeSql(upsertSql);

        //如果对Phoenix中的维度表进行了更新操作，将缓存的数据从Redis中删除掉
        if("update".equals(type)){
            DimUtil.delCached(tableName,jsonObj.getString("id"));
        }

    }
}
